package com.ustc.hewei.nettyrpc2_0.common.config;

import com.ustc.hewei.nettyrpc2_0.common.annotation.AutowireService;
import com.ustc.hewei.nettyrpc2_0.common.annotation.RPCService;
import com.ustc.hewei.nettyrpc2_0.common.enumeration.RpcError;
import com.ustc.hewei.nettyrpc2_0.common.exception.RpcException;
import com.ustc.hewei.nettyrpc2_0.common.factory.ThreadPoolFactory;
import com.ustc.hewei.nettyrpc2_0.common.hook.ShutdownHook;
import com.ustc.hewei.nettyrpc2_0.common.warp.ZkNode;
import com.ustc.hewei.nettyrpc2_0.netty.client.ClientProxy;
import com.ustc.hewei.nettyrpc2_0.netty.client.manager.HandlerManager;
import com.ustc.hewei.nettyrpc2_0.netty.server.net.RpcServer;
import com.ustc.hewei.nettyrpc2_0.zookeeper.register.ServiceRegister;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.x.discovery.ServiceCache;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.ServiceCacheListener;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.annotation.AnnotationUtils;

import java.beans.IntrospectionException;
import java.beans.PropertyDescriptor;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.lang.reflect.AnnotatedElement;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @author hewei
 * @version 1.0
 * @description: 注解扫描类
 * @date 2022/9/16 20:29
 */

@Slf4j
public class DefaultRpcProcessor implements ApplicationListener<ContextRefreshedEvent> {
    private final ServiceRegister serviceRegister;
    private final RpcServer rpcServer;
    private final ClientProxy clientProxy;
    private final HashSet<String> serviceNameList;
    private final ServiceDiscovery<ZkNode> serviceDiscovery;
    private final HandlerManager handlerManager;
    // 是否需要异步更新节点缓存呢？
    //private final ExecutorService executorService;

    public DefaultRpcProcessor(ServiceRegister serviceRegister, RpcServer rpcServer,
                               ClientProxy clientProxy, ServiceDiscovery<ZkNode> serviceDiscovery,
                               HandlerManager handlerManager) {
        this.rpcServer = rpcServer;
        this.clientProxy = clientProxy;
        this.serviceRegister = serviceRegister;
        serviceNameList = new HashSet<>();
        this.serviceDiscovery = serviceDiscovery;
        this.handlerManager = handlerManager;
        //executorService = ThreadPoolFactory.createDefaultThreadPool("watcherEventHandler");
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() == null) {
            ApplicationContext context = event.getApplicationContext();
            //autowireService(context);
            registerRpcServer(context);
        }
    }

    private void registerRpcServer(ApplicationContext context) {
        // 获得上下文中添加了RPCService注解的类对象
        Map<String, Object> beans = context.getBeansWithAnnotation(RPCService.class);
        if (beans.size() != 0) rpcServer.start();
        // 遍历所有的类对象
        for (Object bean : beans.values()) {
            Class<?> clazz = bean.getClass();
            RPCService rpcService = AnnotationUtils.findAnnotation(clazz, RPCService.class);
            String serviceName = rpcService.value().getName();
            int weight = rpcService.weight();
            serviceRegister.register(bean, serviceName, weight);
        }
    }

    public void autowireService(ApplicationContext context) {
        String[] names = context.getBeanDefinitionNames();
        for (String name : names) {
            Class<?> clazz = context.getType(name);
            if (Objects.isNull(clazz)) continue;
            List<Field> fieldList = new ArrayList<>();
            while (clazz != null) {
                fieldList.addAll(new ArrayList<>(Arrays.asList(clazz.getDeclaredFields())));
                clazz = clazz.getSuperclass();
            }
            Field[] fields = new Field[fieldList.size()];
            fieldList.toArray(fields);
            for (Field field : fields) {
                AutowireService annotation = field.getAnnotation(AutowireService.class);
                if (Objects.isNull(annotation)) continue;
                Object bean = context.getBean(name);
                Class<?> fieldType = field.getType();
                try {
                    PropertyDescriptor pd = new PropertyDescriptor(field.getName(), bean.getClass());
                    Method writeMethod = pd.getWriteMethod();
                    writeMethod.invoke(bean, clientProxy.getProxy(fieldType));
                    serviceNameList.add(fieldType.getName());
                } catch (IntrospectionException | IllegalAccessException | InvocationTargetException e) {
                    // 增加容错性，如果没有setter和getter方法，会尝试将代理对象直接注入到字段中，但是不保证成功
                    if (e instanceof IntrospectionException) {
                        log.info("字段 {} 缺失setter或getter方法，将使用默认方式注入!", field.getName());
                        field.setAccessible(true);
                        try {
                            log.info(bean.getClass().getName());
                            field.set(bean, clientProxy.getProxy(fieldType));
                            serviceNameList.add(fieldType.getName());
                        } catch (IllegalAccessException e1) {
                            e1.printStackTrace();
                        }
                    }
                    e.printStackTrace();
                }
//                Object bean = context.getBean(context.getType(name).getSuperclass());
//                Class<?> fieldType = field.getType();
//                serviceNameList.add(fieldType.getName());
//                field.setAccessible(true);
//                try {
//                    field.set(bean, clientProxy.getProxy(fieldType));
//                } catch (IllegalAccessException e) {
//                    e.printStackTrace();
//                }
            }
        }
        if (!serviceNameList.isEmpty()) ShutdownHook.getShutdownHook().addShutdownHook();
        for (String serviceName : serviceNameList) {
            ServiceCache<ZkNode> watcher = serviceDiscovery.serviceCacheBuilder().name(serviceName).build();
            try {
                watcher.addListener(new ServiceCacheListener() {
                    @Override
                    public void cacheChanged() {
                        log.info("服务 {} 发生变动", serviceName);
                        // 异步更新可能会导致顺序错位
                        //executorService.submit(() -> {
                        List<ZkNode> zkNodeList = watcher.getInstances().stream().map(ServiceInstance::getPayload).collect(Collectors.toList());
                        handlerManager.updateCache(serviceName, zkNodeList);
                        log.info("服务 {} 缓存列表已更新", serviceName);
                        //});
                    }

                    @Override
                    public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {

                    }
                });
                watcher.start();
                log.info("服务 {} 监听器已添加", serviceName);
            } catch (Exception e) {
                log.error("监听器添加异常");
                try {
                    watcher.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
                throw new RuntimeException(e);
            }
            for (ServiceInstance<ZkNode> instance : watcher.getInstances()) {
                handlerManager.connectToProvide(instance.getPayload());
            }
        }
    }
}